package org.xiaoleng.cdm.connections;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.xiaoleng.cdm.common.CdmConstants;
import org.xiaoleng.cdm.datasources.Datasource;
import org.xiaoleng.cdm.datasources.RedisDatasource;
import redis.clients.jedis.*;

import java.io.IOException;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;

@Getter
@Setter
@Slf4j
public class RedisConnection extends DatabaseConnection{

    private String name;

    private Jedis jedis;

    private RedisDatasource redisDatasource;

    public RedisConnection(Datasource datasource) {
        super(datasource);
        redisDatasource = (RedisDatasource) datasource;
        jedis = new Jedis(redisDatasource.getAddress(), redisDatasource.getPort());
        //jedis.auth(connectionConfiguration.getPassword());
        jedis.select(1);
        log.info("connect redis successfully.");
    }

    @Override
    public int read(Function<List<JSONObject>, Integer> function) {
        int count = 0;
        // make(); redis 插入数据
        Pipeline pipeline = jedis.pipelined();
        int cursor = 0; // 初始游标
        ScanParams scanParams = new ScanParams().match(redisDatasource.getPattern()).count(CdmConstants.BATCH_SIZE);
        do {
            ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
            cursor = scanResult.getCursor();// 更新游标
            List<String> keys = scanResult.getResult();
            for (String key : keys) {
                pipeline.get(key);
            }
            List<Object> responses = pipeline.syncAndReturnAll();
            List<JSONObject> jsons = responses.stream().map(o -> JSONObject.parseObject(String.valueOf(o)))
                .collect(Collectors.toList());
            List<List<JSONObject>> partitions = Lists.partition(jsons, CdmConstants.BATCH_SIZE);
            for (List<JSONObject> partition : partitions) {
                function.apply(partition);
            }
            count = count + jsons.size();
            jsons.clear();
        } while (cursor != 0); // 当游标为 0 时，表示迭代完成
        try {
            if (pipeline != null) {
                pipeline.close();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return count;
    }

    @Override
    public int write(List<JSONObject> jsons) {
        int count = 0;
        return count;
    }

    @Override
    public void close() {
        if (jedis != null) {
            jedis.close();
            log.info("connection close successfully.");
        }
    }

    private void make() {
        Pipeline pipeline = jedis.pipelined();
        for (int j = 0; j < 100; j++) {
            for (int i = 1; i <= 10000; i++) {
                pipeline.set(j + "-" + i, JSONObject.of("name","n" + i, "age", i).toString());
                System.out.println(i);
            }
            pipeline.sync();
            try {
                if (pipeline != null) {
                    pipeline.close();
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

    }

}
